package org.niit.service

import org.apache.spark.streaming.dstream.DStream
import org.niit.bean.Answer

class EDUDataService {

  def dataAnalysis(answer:DStream[Answer]): Unit ={

    //获取Kafka数据

    hotQuestionTop10(answer)
    gradeQuestionTop10(answer)
    hotQuestionWithSubjectTop10(answer)
    studentMinScoreWithQuestionTop10(answer)
    subjectWithQuestionTop10(answer)
    studentMinScoreWithTop10Question(answer)


  }
  //需求一：统计top10热点题
  private def hotQuestionTop10(answer:DStream[Answer]): Unit ={
    //将结果再控制台打印即可
    //1.  (题目id，1)
    val mapDS =  answer.map(data=>{
      (data.question_id,1)
    })
    //2.(题目id_1，1)  (题目id_1，1)  (题目id_1，1)  (题目id_1，1)
    // (题目id_1，4)  (题目id_2，4)  (题目id_3，4)
    val reduceData = mapDS.reduceByKey(_ + _)
    //3.根据Key进行分组

    //4.排序并输出
    reduceData.foreachRDD(rdd=>{
      // 第一个_ : (题目id_1，4)  _2: 4 默认升序
      val sortRDD = rdd.sortBy(_._2, false)
      val top10 = sortRDD.take(10)
      println("----统计top10热点题----")
      top10.foreach(println)

    })

  }

  //需求二：统计top10答题活跃年级
  private def gradeQuestionTop10(answer:DStream[Answer]): Unit ={
    //将结果再控制台打印即可
    //将结果再控制台打印即可
    //1.  (题目id，1)
    val mapDS =  answer.map(data=>{
      (data.grade_id,1)
    })
    //2.(题目id_1，1)  (题目id_1，1)  (题目id_1，1)  (题目id_1，1)
    // (题目id_1，4)  (题目id_2，4)  (题目id_3，4)
    val reduceData = mapDS.reduceByKey(_ + _)
    //3.排序并输出
    reduceData.foreachRDD(rdd=>{
      // 第一个_ : (题目id_1，4)  _2: 4 默认升序
      val sortRDD = rdd.sortBy(_._2, false)
      val top10 = sortRDD.take(10)
      println("----统计top10答题活跃年级----")
      top10.foreach(println)

    })

    /*
    reduceByKey 和 groupByKey
    reduceByKey:将值进行相加
    groupByKey:将值进行拼接
    ("北海",300)   ("北海",400)  ("北海",200)  ("北海",210)   ("北海",180)
    ==>reduceByKey  : ("北海",1290) 近5年得GDP汇总
    ==> groupByKey: ("北海",[300,400,200,210.180]) 列出近五年中每一年GDP数据
     */

  }

  //需求三：统计top10热点题并带上所属科目  (题目ID_976,科目,22)
  private def hotQuestionWithSubjectTop10(answer:DStream[Answer]): Unit ={
    //将结果再控制台打印即可
    //((题目，科目),次数)  ===  (题目，(科目，次数))
    val mapDS =  answer.map(data=>{
      ((data.question_id,data.subject_id),1 )

      //(data.question_id,(data.subject_id,1))
    })
    //(广西，(南宁，800))  (广西，(北海，300))   (南宁，800))+(北海，300) = (南宁+北海，1100)
    //key一模一样  value 数据类型一样
    val reduceDS =  mapDS.reduceByKey(_+_)
    reduceDS.foreachRDD(rdd=>{
      val sortRDD = rdd.sortBy(_._2,false)
      val top10 = sortRDD.take(10)
      println("----统计top10答题活跃年级----")
      top10.foreach(println)
    })

  }

  //需求四：统计Top10学生最低得分的成绩，并带上题目
  private def studentMinScoreWithQuestionTop10(answer:DStream[Answer]): Unit ={
    //将结果再控制台打印即可
    //key (学生，题目)  value 分数
    val mapDS = answer.map(data=>{
      ( (data.student_id,data.question_id),data.score)
    })

    mapDS.foreachRDD(rdd=>{
      val sortRDD =  rdd.sortBy(_._2,true)
      val top10 =  sortRDD.take(10)
      println("----统计Top10学生最低得分的成绩，并带上题目----")
      top10.foreach(println)
    })

  }
  //需求五：实时 统计每个科目的Top10热点题
  def subjectWithQuestionTop10(answer:DStream[Answer]): Unit ={
    //(科目1，（题目1，20），（题目2，18），.....(题目10，5))
    //key:(科目,题目)    value：次数
    val mapDS =  answer.map(data=>{
      ( (data.subject_id,data.question_id) ,1)
    })
    // ( (科目,题目),10)
    val reduceDS =  mapDS.reduceByKey(_ + _)
    //( (科目,题目),10)  ===> ( 科目,(题目,10) )
    val newMapDS =  reduceDS.map{
      case ( (subject,question),count )=>{//如果符合 ( (科目,题目),10) 格式
        (subject,(question,count))  // 转换为 ( 科目,(题目,10) )
      }
    }
    //groupByKey 根据key进行合并  /(科目1，（题目1，20），（题目2，18），.....(题目………………，5))
    val groupDS =  newMapDS.groupByKey()
    groupDS.foreachRDD(rdd=>{
      val top10 = rdd.mapValues(iter=>{ //iter  （题目1，20）
        val sort =  iter.toList.sortBy(_._2).reverse
        sort.take(10) //获取前10并返回
      })
      println("----实时统计每个科目的Top10热点题----")
      top10.foreach(println)
    })
  }
  //需求六：统计每个学生得分最低的前10道题，并带上所属题目 （学生 ，[(题目1，成绩),(题目2，成绩),(题目3，成绩),………(题目10，成绩), ]）
  def studentMinScoreWithTop10Question(answer:DStream[Answer]): Unit ={

    val mapDS = answer.map(data => { // key （学生，题目） value:成绩
      (data.student_id,(data.question_id,data.score))
    })

    val groupDS = mapDS.groupByKey()//（学生 ，[(题目1，成绩),(题目2，成绩),(题目3，成绩),………(题目……，成绩), ]）
    groupDS.foreachRDD(rdd=>{
      val top10 =  rdd.mapValues(iter=>{//(题目1，成绩)
        val sort =  iter.toList.sortBy(_._2)//sort排序 升序 小--->大
        sort.take(10)
      })
      println("----统计每个学生得分最低的前10道题，并带上所属题目----")
      top10.foreach(println)

    })

  }
}
